package com.greenhouse.mqtt;

import com.greenhouse.config.MqttConfig;
import com.greenhouse.service.DataService;
import com.greenhouse.service.impl.DataServiceImpl;
import jakarta.annotation.PostConstruct;
import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Service
public class MqttService {

    @Autowired
    private MqttConfig mqttConfig;
    @Autowired
    private DataService dataService;
    private static final Logger logger = LoggerFactory.getLogger(MqttService.class);
    private MqttClient mqttClient;

    // 标志位，用于控制是否开始订阅
    private boolean isSubscribed = false;

    // 初始化MQTT客户端
    @PostConstruct
    public void init() throws MqttException {
        mqttClient = mqttConfig.mqttClient();
        setupCallbacks();
    }

    // 设置回调函数
    private void setupCallbacks() {
        mqttClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                 logger.info("Connection to MQTT broker lost!");
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                if (isSubscribed) {
                    logger.info("Received message on topic: " + topic);
                    processIncomingData(topic, message);
                }
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                System.out.println("Delivery complete for message with token: " + token.getMessageId());
            }
        });
    }

    // 订阅所有必要的主题
    @Scheduled(fixedDelay = 10000) // 延迟10秒后订阅
    public void subscribeToTopics() throws MqttException {
        if (!isSubscribed) {
            mqttClient.subscribe(mqttConfig.getSensorsTopic(), mqttConfig.getQos());
            mqttClient.subscribe(mqttConfig.getImagesTopic(), mqttConfig.getQos());
            mqttClient.subscribe(mqttConfig.getVideoTopic(), mqttConfig.getQos());
            isSubscribed = true; // 设置标志位
            System.out.println("Subscribed to topics after delay.");
        }
    }

    // 处理传入的数据
    private void processIncomingData(String topic, MqttMessage message) {
        //处理topic字符串按/分开
        String[] parts = topic.split("/");
        switch (parts[0]){
            case "sensors":
                dataService.insertData(parts,message);
                break;
            case "images":
                break;
            case "video":
                break;
            default:
                break;

        }
    }

    public void sendData(String topic, byte[] payload) throws MqttException {
        MqttMessage message = new MqttMessage(payload);
        message.setQos(mqttConfig.getQos());
        mqttClient.publish(topic, message);
    }

    private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    @PostConstruct
    public void startHeartbeat() {
        Runnable heartbeatTask = () -> {
            try {
                sendData("heartbeat", "Device is alive".getBytes());
            } catch (MqttException e) {
                e.printStackTrace();
            }
        };
        scheduler.scheduleAtFixedRate(heartbeatTask, 0, 1800, TimeUnit.SECONDS);
    }
}